package org.onion.pattern.event.springboot.rocketmq.consume.listener;


import lombok.Data;
import lombok.Getter;
import lombok.Setter;
import lombok.SneakyThrows;
import lombok.extern.slf4j.Slf4j;
import org.apache.commons.collections.MapUtils;
import org.apache.commons.lang3.StringUtils;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.remoting.protocol.NamespaceUtil;
import org.onion.pattern.event.consumer.EventConsumeHandler;
import org.onion.pattern.event.consumer.ExternalEventConverter;
import org.onion.pattern.event.consumer.ExternalEventInfo;
import org.onion.pattern.event.exception.EventConsumeIdempotentException;
import org.onion.pattern.event.spring.io.EventCodecs;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMQ;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMQMessageProperties;
import org.onion.pattern.event.springboot.rocketmq.config.RocketMessageMetadata;
import org.onion.pattern.event.springboot.rocketmq.consume.TopicAndEventTypeMapping;
import org.onion.pattern.event.springboot.rocketmq.exception.RocketMQEventConsumeException;
import org.springframework.lang.Nullable;

import java.time.Instant;
import java.time.LocalDateTime;
import java.time.ZoneId;
import java.util.*;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Data
abstract class _MessageListener {

    protected EventConsumeHandler          consumeHandler;
    protected DefaultMQPushConsumer        consumer;
    protected TopicAndEventTypeMapping topicAndEventTypeMapping;
    protected EventCodecs                  eventCodecs;
    @Setter
    @Getter
    protected List<ExternalEventConverter> converters = new ArrayList<>();
    private   Map<String, Class<?>>        classCache = new ConcurrentHashMap<>();

    @SneakyThrows
    protected ExternalEventInfo deserialize(MessageExt messageExt, @Nullable Class<?> eventType, RocketMessageMetadata metadata) {
        final String deserializeType = messageExt.getProperty(RocketMQMessageProperties.PAYLOAD_SERIALIZER_TYPE);
        final String className = messageExt.getProperty(RocketMQMessageProperties.PAYLOAD_CLASS_NAME);
        final String eventTime = messageExt.getProperty(RocketMQMessageProperties.EVENT_TIME);
        if (StringUtils.isNotBlank(className)) {
            eventType = classCache.get(className);
            if (eventType == null) {
                eventType = Class.forName(className);
                classCache.put(className, eventType);
            }
        }

        Object payload = eventCodecs.deserialize(deserializeType)
                .deserialize(messageExt.getBody(), eventType);

        ExternalEventInfo externalEventInfo = new ExternalEventInfo();
        externalEventInfo.setEventKey(metadata.getMessageKey());
        try {
            externalEventInfo.setTs(LocalDateTime.ofInstant(Instant.ofEpochMilli(Long.parseLong(eventTime)), ZoneId.systemDefault()));
        } catch (NumberFormatException e) {
            log.warn("",e);
            externalEventInfo.setTs(metadata.getStoreTime());
        }
        externalEventInfo.setPayload(payload);
        externalEventInfo.setSource(RocketMQ.CHANNEL_NAME);
        return externalEventInfo;
    }


    protected void handleMessage(MessageExt messageExt) throws RocketMQEventConsumeException {

        RocketMessageMetadata rocketMessageMetadata = new RocketMessageMetadata();
        String consumerGroup = NamespaceUtil.wrapNamespace(consumer.getNamespace(), consumer.getConsumerGroup());
        String topic = NamespaceUtil.wrapNamespace(consumer.getNamespace(), messageExt.getTopic());

        rocketMessageMetadata.setConsumerGroup(consumerGroup);
        rocketMessageMetadata.setTopic(topic);
        rocketMessageMetadata.setSubTopic(messageExt.getTags());
        rocketMessageMetadata.setMessageId(messageExt.getMsgId());
        rocketMessageMetadata.setMessageKey(messageExt.getKeys());
        rocketMessageMetadata.setQueueId(messageExt.getQueueId());
        rocketMessageMetadata.setMessageOffset(messageExt.getCommitLogOffset());
        rocketMessageMetadata.setReconsumeTimes(messageExt.getReconsumeTimes());
        rocketMessageMetadata.setBroker(Optional.ofNullable(messageExt.getBrokerName()).orElseGet(() -> "unknown"));
        rocketMessageMetadata.setBornTimestamp(messageExt.getBornTimestamp());
        Properties properties = MapUtils.toProperties(messageExt.getProperties());
        rocketMessageMetadata.setProperties(properties);
        rocketMessageMetadata.setStoreTime(LocalDateTime.ofInstant(Instant.ofEpochMilli(messageExt.getStoreTimestamp()), ZoneId.systemDefault()));

        final Class<?> eventType = topicAndEventTypeMapping.getEventType(messageExt.getTopic(), messageExt.getTags());
        ExternalEventInfo externalEventInfo = deserialize(messageExt, eventType, rocketMessageMetadata);

        if (externalEventInfo == null) {
            log.error("【事件消费】消息消费失败，无法获取转换事件报文, group->{}, topic->{}, message_key->{}",
                    consumerGroup, messageExt.getTopic(),
                    messageExt.getKeys());
        }

        try {
            this.consumeHandler.handle(externalEventInfo, rocketMessageMetadata);
        } catch (EventConsumeIdempotentException e) {
            log.warn("【事件消费】 重复消费 {}", e.getMessage());
        } catch (Exception e) {
            throw new RocketMQEventConsumeException(e);
        }
    }


}
